package com.xiaomi.config;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RocketMQConfig {

    @Bean
    public DefaultMQProducer producer() throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer("producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();
        return producer;
    }

    @Bean
    public DefaultMQPushConsumer consumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        //订阅 TestTopic 主题下的所有消息
        consumer.subscribe("TestTopic", "*");
        // 注册消息监听器，使用并发消费模式处理接收到的消息
        consumer.registerMessageListener((MessageListenerConcurrently) (messageExts, consumeConcurrentlyContext) -> {
            for (MessageExt messageExt : messageExts) {
                // 打印接收到的消息内容
                System.out.println(new String(messageExt.getBody()));
            }
            // 表示消息消费成功，告知 RocketMQ 服务端
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        return consumer;
    }
}
